package com.qiuling.iot.config;


import com.qiuling.iot.rabbit.RabbitConfirmCallBack;
import com.qiuling.iot.rabbit.RabbitReturnCallback;
import com.qiuling.iot.rabbit.MyMessageConverter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @Author：zhangqiang
 * @Description：
 * @Date：Created in 0:04 2020/3/6
 * @Modified By：
 */
@Configuration
@Slf4j
public class RabbitConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host}")
    private String virtualHost;

    @Value("${spring.rabbitmq.listener.concurrency}")
    private int concurrency;

    @Value("${spring.rabbitmq.listener.max-concurrency}")
    private int maxConcurrency;

    @Value("${spring.rabbitmq.listener.prefetch}")
    private int prefetch;

    @Autowired
    private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }


    /**
     * 处理消费监听配置
     * 单一消费者
     *
     * @return
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        factory.setMessageConverter(new MyMessageConverter());
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(1);
        factory.setTxSize(1);
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        return factory;
    }

    /**
     * 多消费者
     *
     * @return
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factoryConfigurer.configure(factory, connectionFactory());
        factory.setMessageConverter(new MyMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.NONE);
        factory.setConcurrentConsumers(concurrency);
        factory.setMaxConcurrentConsumers(maxConcurrency);
        factory.setPrefetchCount(prefetch);
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMessageConverter(new MyMessageConverter());
        template.setConfirmCallback(new RabbitConfirmCallBack());
        template.setReturnCallback(new RabbitReturnCallback());
        log.info("初始化rabbitmqTemplate:{}", template);
        return template;
    }


    /**
     * Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,
     * Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。
     * Queue:消息的载体,每个消息都会被投到一个或多个队列。
     * Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.
     * Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
     * vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。
     * Producer:消息生产者,就是投递消息的程序.
     * Consumer:消息消费者,就是接受消息的程序.
     * Channel:消息通道,在客户端的每个连接里,可建立多个channel.
     */
    //设置路由
    public static final String IOT_EXCHANGE = "iot.exchange";
    //设置队列
    public static final String BACK_QUEUE = "back.queue";
    public static final String DATA_QUEUE = "data.queue";
    public static final String SMS_QUEUE = "sms.queue";
    //设置routingKey
    public static final String BACK_ROUTING_KEY = "back.routing.key";
    public static final String DATA_ROUTING_KEY = "data.routing.key";
    public static final String SMS_ROUTING_KEY = "sms.routing.key";

    /**
     * 针对消费者配置
     * 1. 设置交换机类型
     * 2. 将队列绑定到交换机
     * FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
     * HeadersExchange ：通过添加属性key-value匹配
     * DirectExchange:按照routingkey分发到指定队列
     * TopicExchange:多关键字匹配
     */
    @Bean
    public DirectExchange defaultExchange() {
        return new DirectExchange(IOT_EXCHANGE);
    }

    @Bean
    public Queue backQueue() {
        return new Queue(BACK_QUEUE);
    }

    @Bean
    public Queue dataQueue() {
        //队列持久
        return new Queue(DATA_QUEUE, true);
    }

    @Bean
    public Queue SMSQueue() {
        //队列持久
        return new Queue(SMS_QUEUE, true);
    }

    /**
     * direct模式
     * 消息中的路由键（routing key）如果和 Binding 中的 binding key 一致，
     * 交换器就将消息发到对应的队列中。路由键与队列名完全匹配
     *
     * @return
     */
    @Bean
    public Binding bindingCallback() {
        return BindingBuilder
                // 设置queue
                .bind(backQueue())
                // 绑定交换机
                .to(defaultExchange())
                // 设置routingKey
                .with(BACK_ROUTING_KEY);
    }

    @Bean
    public Binding bindingData() {
        return BindingBuilder
                // 设置queue
                .bind(dataQueue())
                // 绑定交换机
                .to(defaultExchange())
                // 设置routingKey
                .with(DATA_ROUTING_KEY);
    }

    @Bean
    public Binding bindingSMS() {
        return BindingBuilder
                // 设置queue
                .bind(SMSQueue())
                // 绑定交换机
                .to(defaultExchange())
                // 设置routingKey
                .with(SMS_ROUTING_KEY);
    }

}
